package com.techsun.datanow.sync

import com.microsoft.azure.cosmosdb.spark.config.Config
import com.microsoft.azure.cosmosdb.spark.schema._
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{StructField, _}

/**
 *
 * @author Song
 * @date 2020/9/22
 */
class Sync_ThirdinfoSync extends TDataSync {
  /**
   * 同步工作方法
   *
   * @param spark             spark 会话对象
   * @param cosmosDbEndpoint  CosmosDB Endpoint
   * @param cosmosDbMasterkey CosmosDB Endpoint 对应的密钥
   * @param kuduMaster        kudu Master 节点
   * @return 同步成功或失败
   */
  override def sync(spark: SparkSession, cosmosDbEndpoint: String, cosmosDbMasterkey: String, kuduMaster: String, beginTime: Long, endTime: Long): Boolean = {
    import spark.implicits._
    try {
      Logger.getLogger("org").setLevel(Level.ERROR)
      Logger.getLogger("com").setLevel(Level.ERROR)
      val kuduContext = new KuduContext(kuduMaster, spark.sparkContext)
      val config = Config(Map("Endpoint" -> cosmosDbEndpoint,
        "Masterkey" -> cosmosDbMasterkey,
        "Database" -> "VF_Vans_CN",
        "preferredRegions" -> "China East 2;",
        "Collection" -> "TsrMemberInfo",
        "query_custom" ->
          s"""SELECT c.id, c.tp, c.createTime, c.updateTime
             |FROM c
             | where c._ts >= ${beginTime}""".stripMargin))


      val schema = StructType(
        List(
          StructField("id", StringType, true),
          StructField("tp", ArrayType(StructType(List(
            StructField("id", StringType, true),
            StructField("store", StringType, true),
            StructField("type", IntegerType, true),
            StructField("thirdId", StringType, true),
            StructField("nickName", StringType, true),
            StructField("isBind", IntegerType, true),
            StructField("bindTime", TimestampType, true),
            StructField("unbindTime", TimestampType, true),
            StructField("openId1", StringType, true),
            StructField("openId2", StringType, true),
            StructField("openId3", StringType, true),
            StructField("openId4", StringType, true),
            StructField("openId5", StringType, true),
            StructField("openId6", StringType, true),
            StructField("openId7", StringType, true),
            StructField("externalCampaignCode", StringType, true),
            StructField("createTime", TimestampType, true),
            StructField("updateTime", TimestampType, true)
          )), true), true),
          StructField("createTime", TimestampType, true),
          StructField("updateTime", TimestampType, true)
        )
      )

      val memberInfo = spark.sqlContext.read.schema(schema).cosmosDB(config)

      val thirdInfo = memberInfo.select($"id", explode($"tp"), $"createTime", $"updateTime").toDF("id", "tp", "createTime", "updateTime").select(
        $"id".alias("memberid"), $"tp.id".alias("tpid"), $"tp.store".alias("store"), $"tp.type".alias("type"), $"tp.thirdId".alias("thirdid"), $"tp.nickName".alias("nickname")
        , $"tp.isBind".alias("isbind"), $"tp.bindTime".alias("bindtime"), $"tp.unbindTime".alias("unbindtime")
        , $"tp.openId1".alias("openid1"), $"tp.openId2".alias("openid2"), $"tp.openId3".alias("openid3"), $"tp.openId4".alias("openid4")
        , $"tp.openId5".alias("openid5"), $"tp.openId6".alias("openid6"), $"tp.openId7".alias("openid7"), $"tp.externalCampaignCode".alias("externalcampaigncode")
        , $"tp.createTime".alias("tpcreatetime"), $"tp.updateTime".alias("tpupdatetime"), $"createTime".alias("createtime"), $"updateTime".alias("updatetime"))
      val count = thirdInfo.count();
      printf("Sync_TsrThirdInfo query count: %d\n", count)
      kuduContext.upsertRows(thirdInfo, "sh_tsrthirdinfo")

      return true
    } catch {
      case ex: Exception => {
        println("Sync_TsrThirdInfo exception: " + ex.getMessage)
        return false
      }
    }
  }
}
